package protocol

import (
	"context"
	"sync"

	"github.com/golang/protobuf/proto"
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/libp2p/go-libp2p-core/protocol"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/p2p/common"
	"hundsun.com/hsl/hschain/p2p/impl/base"
	"hundsun.com/hsl/hschain/p2p/impl/dht/node"
	"hundsun.com/hsl/hschain/p2p/impl/dht/txpool"
	"hundsun.com/hsl/hschain/store/mq"
)

// P2PEnv ...
type P2PEnv struct {
	logging.Log
	Cfg         *common.P2pConfig
	Ctx         context.Context
	TopicClient mq.Client  // topic类型client
	WorkClient  mq.Client  // work类型client
	P2pNode     *node.Node // p2p节点
	TxPool      txpool.PendingTxPool
}

//SendMessages 用于广播消息或者将消息发给相应策略的节点
func (p *P2PEnv) SendMessages(msg proto.Message, ptl protocol.ID, opts ...base.MessageOption) error {
	msgOpts := base.GetMessageOption(opts)

	filter := p.P2pNode.GetFilter(msgOpts)
	res, err := filter.Filter()
	if err != nil {
		p.Log.Error("sendMessages", "filter error", err)
		return err
	}
	peers := res.([]peer.ID)
	peersRes := make(map[peer.ID]struct{})

	//todo 证书握手通过后加入白名单
	// 白名单过滤
	if len(msgOpts.WhiteList) > 0 {
		for _, v := range peers {
			if _, exist := msgOpts.WhiteList[v.Pretty()]; exist {
				peersRes[v] = struct{}{}
			}
		}
	} else {
		for _, v := range peers {
			peersRes[v] = struct{}{}
		}
	}

	// 黑名单过滤
	for k := range peersRes {
		if _, exist := msgOpts.BlackList[k.Pretty()]; exist {
			delete(peersRes, k)
		}
	}

	// 并行发送消息
	wg := &sync.WaitGroup{}
	for pr := range peersRes {
		wg.Add(1)
		go func(pr peer.ID) {
			defer wg.Done()
			p.P2pNode.SendMessage(msg, ptl, pr)
		}(pr)
	}
	wg.Wait()
	return nil
}
